package com.peng.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

//生产者API使用
public class CustomerProducer {
    public static void main(String[] args) {
        //1 创建Kafka生产者需要的配置信息
         Properties props = new Properties();

         props.put("bootstrap.servers", "localhost:9092");//Kafka集群
         props.put("acks", "all");//应答机制的级别
        // props.put(ProducerConfig.ACKS_CONFIG,"all");//这些参数都可以通过ProducerConfig.XXX 来获得

         props.put("retries", 0);//重试次数，0次表示发送失败了就不管了
         props.put("batch.size", 16384);//批量大小，一次发送的缓存
         props.put("linger.ms", 1);//提交延迟，1ms或者数据量到16k就 提交数据
         props.put("buffer.memory", 33554432);//整个producer的缓存数据为32m
        //KV的序列化类
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("partitioner.class","com.peng.producer.ProducerPartitioner");

         //2 创建生产者来发送数据
         Producer<String, String> producer = new KafkaProducer<String, String>(props);
         //循环发送数据
         for (int i = 0; i < 2; i++)
            producer.send(new ProducerRecord<String, String>("first", Integer.toString(i)), new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if(exception==null){
                        System.out.println((+metadata.partition()+"--"+metadata.offset()));
                    }else{
                        System.out.println("发送失败");
                    }
                }
            });
         //关闭资源
         producer.close();
    }
}
